package demo.hbase.service;

import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.io.IOException;

/**
 * @description description
 * @author zhouxinghuai
 * @date 2018-09-18
 */
@Service
public class HBaseService0 {

    @Autowired
    private Configuration hbaseConf;

    static Connection conn;


    @PostConstruct
    public void createConn() {
        try {
            conn = ConnectionFactory.createConnection(hbaseConf);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    @PreDestroy
    public void closeConn() {
        if (conn != null) {
            try {
                conn.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }


    /**
     * 建表
     * @param tableName 表名
     * @param family 列族
     * @throws IOException
     */
    public void createTable(String tableName, String[] family) throws IOException {
        // 创建表webtable 三个列contents anchor people
        if (StringUtils.isEmpty(tableName)) {
            return;
        }
        if (ArrayUtils.isEmpty(family)) {
            return;
        }

        HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();

        // 表HTableDescriptor 列HColumnDescriptor
        HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName));
        for (String f : family) {
            desc.addFamily(new HColumnDescriptor(f));
        }

        if (admin.tableExists(TableName.valueOf(tableName))) {
            System.out.println("table Exists!");
        } else {
            admin.createTable(desc);
            System.out.println("create table success!");
        }
        admin.close();

    }

    /**
     * 插入（更新）数据
     * @param rowKey
     * @param tableName
     * @param family 列族
     * @param qualifier 列
     * @param value 值
     * @throws IOException
     */
    public void putData(String rowKey, String tableName, byte[] family, byte[] qualifier, byte[] value) throws IOException {
        if (StringUtils.isEmpty(rowKey) || StringUtils.isEmpty(tableName)) {
            return;
        }

        if (ArrayUtils.isEmpty(family) || ArrayUtils.isEmpty(qualifier)) {
            return;
        }

        Put put = new Put(Bytes.toBytes(rowKey));
        Table table = conn.getTable(TableName.valueOf(tableName));
        put.addColumn(family, qualifier, value);
        table.put(put);
        System.out.println("add data Success!");
        table.close();
    }

    /**
     * 遍历指定rowKey的数据
     * @param tableName
     * @param rowKey
     * @return
     */
    public Result listCells(String tableName, String rowKey) throws IOException {
        Get get = new Get(Bytes.toBytes(rowKey));
        Table table = conn.getTable(TableName.valueOf(tableName));
        Result result = table.get(get);

        System.out.println("result.current():" + result.current());

        CellScanner cellScanner = result.cellScanner();

        System.out.println("cellScanner:" + cellScanner.toString());
        System.out.println("cellScanner.current():" + cellScanner.current());

        System.out.println("-----------------------------");

        for (Cell cell : result.listCells()) {
            System.out.println("cell:" + cell.toString());
            System.out.println("family:" + Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()));
            System.out.println("qualifier:" + Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()));
            System.out.println("value:" + Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
            System.out.println("timestamp:" + cell.getTimestamp());
            System.out.println("-----------------------------");
        }

        return result;
    }


    /**
     * 遍历指定rowKey和列的数据
     *
     * @param tableName
     * @param rowKey
     * @return Result 可自定义某种数据封装
     * @throws IOException
     */
    public static Result listCells(String tableName, String rowKey, byte[] family, byte[] qualifier) throws IOException {
        Get get = new Get(Bytes.toBytes(rowKey));
        get.setMaxVersions(Integer.MAX_VALUE);
        Table table = conn.getTable(TableName.valueOf(tableName));
        Result result = table.get(get);

        for (Cell cell : result.getColumnCells(family, qualifier)) {
            System.out.println("cell: " + cell.toString());
            System.out.println("family: " + Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()));
            System.out.println("qualifier: " + Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()));
            // value要根据实际类型转换toString等
            System.out.println("value: " + Bytes.toLong(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
            System.out.println("timestamp: " + cell.getTimestamp());
            System.out.println("***************************************");
        }

        return result;
    }

    /**
     * 获取指定rowKey和列最新版本的数据
     *
     * @param tableName
     * @param rowKey
     * @return Result 可自定义某种数据封装
     * @throws IOException
     */
    public static Result latestCell(String tableName, String rowKey, byte[] family, byte[] qualifier) throws IOException {
        Get get = new Get(Bytes.toBytes(rowKey));
        Table table = conn.getTable(TableName.valueOf(tableName));
        Result result = table.get(get);

        Cell cell = result.getColumnLatestCell(family, qualifier);
        System.out.println("cell: " + cell.toString());
        System.out.println("family: " + Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength()));
        System.out.println("qualifier: " + Bytes.toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength()));
        System.out.println("value: " + Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength()));
        System.out.println("timestamp: " + cell.getTimestamp());

        return result;
    }


}
